package com.carol.bigdata.fuction

import com.alibaba.fastjson.{JSON, JSONObject}
import com.carol.bigdata.utils.TimeUtil
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD

import scala.collection.JavaConverters._
import scala.collection.mutable

object LogParseUtil extends Serializable {

    /**
     *{"@timestamp":"2021-03-31T08:29:59.776Z",
     * "@metadata":{"beat":"filebeat","type":"_doc","version":"7.9.3"},
     * "stream":"stderr",
     * "message":{
     * "request":{"timestamp":"1617179399775","agent":"python-requests/2.23.0","game_id":"2","ip":"172.99.99.99:64592",
     * "time":"1601547392000","event":"CAG_chat",
     * "properties":{"chat_type":"大厅","chat_uid":"","message":"伫"},
     * "type":"action","uid":"a63c90cc3684ad8b0a2176a6a8fe9005"},
     * "type":"hdfs","app":"sdk_event","level":"info","ts":1.6171793997758946e+09,"caller":"log/logger.go:48","msg":"event"},"input":{"type":"docker"},"ecs":{"version":"1.5.0"},"host":{"name":"08aba45f9253"},"agent":{"version":"7.9.3","hostname":"08aba45f9253","ephemeral_id":"965ae48b-e757-422c-936c-c3f61789554d","id":"552e45ea-8fe5-4d94-9587-248ef2c3a4ee","name":"08aba45f9253","type":"filebeat"},"log":{"offset":240,"file":{"path":"/var/lib/docker/containers/24414463c1b5ada969e1012c38d6b48b8b85ca8e91f51f6e69ca279358a8d9fb/24414463c1b5ada969e1012c38d6b48b8b85ca8e91f51f6e69ca279358a8d9fb-json.log"}}}
     */
     
    def filebeat2log(value: String, myAccumulator: MyAccumulator): (String, Map[String, (String, String)]) = {
        val data: JSONObject = JSON.parseObject(value).getJSONObject("message").getJSONObject("request")
        val map: mutable.HashMap[String, (String,String)] = new mutable.HashMap[String, (String,String)]()
        var uid = ""
        var uid_type = ""
        try {
            val jsonKey = data.keySet()
            val iter = jsonKey.iterator()
            uid = data.getString("uid")
            uid_type = data.getString("uid_type")
            val time = data.getString("time")
            val timezone = data.getString("timezone")
            val timeStr = TimeUtil.timestamp2str(time, timezone = timezone)
            // 添加新的gameId到累加器中
            val game_id = data.get("game_id").toString.toInt
            myAccumulator.add(game_id)
            // 解析字段封装到map中，会拆分properties
            while (iter.hasNext) {
                val instance = iter.next()
                var value = ""
                value = if (data.get(instance) == null) "" else data.get(instance).toString.toLowerCase
                if (instance.toLowerCase == "timezone") value = data.get(instance).toString
                if (instance == "properties") {
                    val propMap = data.getJSONObject(instance).asScala
                    for ((k, v) <- propMap) map.put(k, (v.toString,timeStr))
                }
                else {
                    map.put(instance, (value,timeStr))
                }
            }
        }catch {
            case e: NullPointerException => {
                println("缺少关键字段")
            }
        }
        (uid+"-"+uid_type,map.toMap)
    }
    
    // 解析日志格式
    def parseLog(rdd: RDD[ConsumerRecord[String, String]],
                 myAccumulator: MyAccumulator): RDD[(String, Map[String, (String, String)])] = {
        val value = rdd.map(x => {
            val topic: String = x.topic()
            val value: String = x.value()
            if (topic.contains("filebeat")) {
                LogParseUtil.filebeat2log(value,myAccumulator)
            }
            else {
               ("",Map[String, (String,String)]())
            }
        })
        value
    }
}
